package WebserviceChecker;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import DatabaseManaging.JdbcChannelsDao;
import DatabaseManaging.JdbcLanguageDao;
import DatabaseManaging.JdbcProductsDao;
import DatabaseManaging.LanguageChannelDatabaseChecker;
import DatabaseManaging.ParametersDAO;
import DatabaseManaging.ProductDatabaseChecker;

public class CheckWebServiceProcess extends Configured implements Tool{
	
	public Path prodPath;
	public static class LogMapper extends Mapper<Object, Text, Text, Text> {
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//																Mapper															   //							
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	public void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
		String[] record=value.toString().split("\t",2);
		String parameter=record[0];
		context.write(new Text(parameter), new Text(record[1]));
	}
}
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//											Classes to Implement     														   //							
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	public interface CheckWBInterface{
		public ArrayList<String> checkWebservice(ApplicationContext context,Iterable<Text> inputValues);
	}
	
	public static class ProductChecker implements CheckWBInterface{
		@Override
		public ArrayList<String> checkWebservice(ApplicationContext context,Iterable<Text> inputValues) {
			System.out.println("Products");
			ParametersDAO dao3=(JdbcProductsDao) context.getBean("JdbcProductsDao");
		    ProductDatabaseChecker pdDBProd=new ProductDatabaseChecker();
		    pdDBProd.setJdbc(dao3);
		    ProductWbChecker pdCheckerProd=new ProductWbChecker(pdDBProd);
		    ArrayList<String> errorsList=pdCheckerProd.checkWebService(inputValues);
		    return errorsList;
		  }	
	}
	
	public static class LanguageChecker implements CheckWBInterface{
		@Override
		public ArrayList<String> checkWebservice(ApplicationContext context,Iterable<Text> inputValues) {
		    JdbcLanguageDao dao=(JdbcLanguageDao) context.getBean("JdbcLanguagesDao");
		    LanguageChannelDatabaseChecker langchecker=new LanguageChannelDatabaseChecker();
		    langchecker.setJdbc(dao);
			ChannelLanguageChecker pdCheckerLangs=new ChannelLanguageChecker(langchecker);
			pdCheckerLangs.setParameter("languages");
			ArrayList<String> errorsList=pdCheckerLangs.checkWebService(inputValues);
			return errorsList;
		  }	
	}
	public static class ChannelChecker implements CheckWBInterface{
		@Override
		public ArrayList<String> checkWebservice(ApplicationContext context,Iterable<Text> inputValues) {
		    JdbcChannelsDao dao=(JdbcChannelsDao) context.getBean("JdbcChannelsDao");
		    LanguageChannelDatabaseChecker channchecker=new LanguageChannelDatabaseChecker();
		    channchecker.setJdbc(dao);
			ChannelLanguageChecker pdCheckerChannels=new ChannelLanguageChecker(channchecker);
			pdCheckerChannels.setParameter("channels");
			ArrayList<String> errorsList=pdCheckerChannels.checkWebService(inputValues);
			return errorsList;
		  }	
	}
	
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//															Reducer															   //							
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////	
public static class LogReducer extends
		Reducer<Text, Text, Text, Text> {
	
	CheckWBInterface checker;
	
	public CheckWBInterface getChecker() {
		return checker;
	}

	public void setChecker(CheckWBInterface checker) {
		this.checker = checker;
	}
	
	public void reduce(Text key, Iterable<Text> values,
			Context context) throws IOException, InterruptedException {
	    ApplicationContext cont = new ClassPathXmlApplicationContext("application-context-persist.xml");
		System.out.println("KEY:"+key.toString());
	    if("products".equals(key.toString())){
	    	System.out.println("--------------List products: ------------------");
	    	this.setChecker(new ProductChecker());
		}else if("languages".equals(key.toString())){
	    	System.out.println("--------------List Language: ------------------");
			this.setChecker(new LanguageChecker());
		}else if("channels".equals(key.toString())){
	    	System.out.println("--------------List Channels: ------------------");
			this.setChecker(new ChannelChecker());
		}
	    System.out.println(values);
    	ArrayList<String> errorsList=checker.checkWebservice(cont, values);
    	for(int i=0;i<errorsList.size();i++){
    		context.write(key, new Text(errorsList.get(i)));
    	}
	}
}

/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//															Main															   //							
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////	



public int run(String[] args) throws Exception { 
	Configuration conf = getConf();
	
	Job job = new Job(conf, "ParseLogs");
	job.setJarByClass(CheckWebServiceProcess.class);

	job.setMapperClass(LogMapper.class);
	job.setReducerClass(LogReducer.class);
	
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(Text.class);

    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]) );

	System.exit(job.waitForCompletion(true)?0:1);
	
	return 0;

}

public static void main(String[] args) throws Exception {
	int res = ToolRunner.run(new Configuration(), new CheckWebServiceProcess(), args);
	System.exit(res);
	}

}
